package org.example;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import java.io.UnsupportedEncodingException;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;

/**
 * @author : libo@juzishuke.com
 * @Title :
 * @Description :
 * @date : 2024/3/14 13:34
 * @since :1.0
 */
@Slf4j
public class ConsumerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
        properties.put(PropertyKeyConst.GROUP_ID, "GID_LEE_GROUP");

        /**
         * 如果是使用公网接入点访问，则必须设置AccessKey和SecretKey，里面填写实例的用户名和密码。实例用户名和密码在控制台实例详情页面获取。
         * 注意！！！这里填写的不是阿里云账号的AccessKey ID和AccessKey Secret，请务必区分开。
         * 如果是在阿里云ECS内网访问，则无需配置，服务端会根据内网VPC信息智能获取。
         * 如果实例类型为Serverlesss实例，则不管是公网访问还是内网访问都必须设置实例的用户名密码。
         */
        // 设置为消息队列RocketMQ版控制台实例详情页的实例用户名。
        properties.put(PropertyKeyConst.AccessKey, "42ae72c3d6fd3aass64d31149b677238");
        // 设置为消息队列RocketMQ版控制台实例详情页的实例密码。
        properties.put(PropertyKeyConst.SecretKey, "22881f250daa21mk8075fb2");
        //注意！！！使用ONS SDK访问RocketMQ 5.x实例时，InstanceID属性不需要设置，否则会导致失败。

        // 设置为您从消息队列RocketMQ版控制台获取的接入点，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
        // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "lt-mq-server.juzishuke.com:9876");
        // 顺序消息消费失败进行重试前的等待时间，单位（毫秒），取值范围：10毫秒~30000毫秒。
        properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
        // 消息消费失败时的最大重试次数。
        properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");

        // 在订阅消息前，必须调用start方法来启动Consumer，只需调用一次即可。
        OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);

        consumer.subscribe(
                // 设置为您在消息队列RocketMQ版控制台上创建的Topic。
                "TOPIC_LEE_TEST",
                // 订阅指定Topic下的Tags：
                // 1. * 表示订阅所有消息。
                // 2. TagA || TagB || TagC表示订阅TagA或TagB或TagC的消息。
                "*", new MessageOrderListener() {
                    /**
                     * 1. 消息消费处理失败或者处理出现异常，返回OrderAction.Suspend。
                     * 2. 消息处理成功，返回OrderAction.Success。
                     */
                    @Override
                    public OrderAction consume(Message message, ConsumeOrderContext context) {
                        try {
                            if (message.getShardingKey().equals("biz_5")) {
                                Thread.sleep(1000);
                                log.info(new String(message.getBody(), "utf-8"));
                            } else {
                                log.info(new String(message.getBody(), "utf-8"));
                            }
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        } catch (UnsupportedEncodingException e) {
                            throw new RuntimeException(e);
                        }
                        return OrderAction.Success;
                    }
                });

        consumer.start();
    }
}